/**
 * 
 */

package com.wutianyi.queue.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import com.wutianyi.queue.Task;
import com.wutianyi.queue.TaskQueue;
import com.wutianyi.queue.dao.TaskDao;

/**
 * 将超过大小限制的task保存到数据库中
 * 里面会建立两个queue作为缓存
 * 
 * @author wutianyi
 * 
 */
public class DBStoreQueue<T extends Task> implements TaskQueue<T> {

	/**
	 * 
	 */
	private static final int		MAX_BUFFER_SIZE	= 1000;

	/**
	 * 如果超过最大数，每次存入数据库的最大任务数
	 */
	private static final int		MAX_STORE_SIZE	= 500;

	private AtomicInteger			queueSize		= new AtomicInteger(0);

	private int						bufferSize;
	/**
	 * 每次存入或获取数据库中任务的数目
	 * 这个数最好保持是bufferSize/2
	 */
	private int						storeSize;
	private TaskDao					taskDao;
	private ReentrantLock			lock			= new ReentrantLock();

	private LinkedBlockingQueue<T>	buffer;

	/**
	 * 定义单线程模式的存储线程
	 */
	private ExecutorService			executor;
	/**
	 * 具体的执行任务
	 */
	private StoreTask				storeTask;

	public DBStoreQueue(int bufferSize, int storeSize) {

		this.bufferSize = bufferSize;
		this.storeSize = storeSize;
		buffer = new LinkedBlockingQueue<T>();
		executor =
				new ThreadPoolExecutor(1, 1, Long.MAX_VALUE, TimeUnit.NANOSECONDS,
						new LinkedBlockingQueue<Runnable>(1),
						new ThreadPoolExecutor.DiscardPolicy());
		storeTask = new StoreTask();
	}

	public DBStoreQueue() {

		this(MAX_BUFFER_SIZE, MAX_STORE_SIZE);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#batchAddTask(java.util.List)
	 */
	@Override
	public void batchAddTask(List<T> tasks) {

		// TODO Auto-generated method stub

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#drainTo(java.util.Collection, int)
	 */
	@Override
	public int drainTo(Collection<T> c, int size) {

		// TODO Auto-generated method stub
		return 0;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#drainTo(int)
	 */
	@Override
	public List<T> drainTo(int size) {

		// TODO Auto-generated method stub
		return null;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#offer(com.wutianyi.queue.Task)
	 */
	@Override
	public boolean offer(T t) {

		boolean result = buffer.offer(t);
		if(result) {
			queueSize.getAndIncrement();
		}

		if(buffer.size() >= bufferSize) {
			executor.execute(storeTask);
		}

		return result;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#poll()
	 */
	@Override
	public T poll() {

		if(0 == buffer.size()) {
			fetchStasks();
		}
		T t = buffer.poll();
		if(null != t) {
			queueSize.getAndDecrement();
		}
		return t;
	}

	private void fetchStasks() {

		try {
			lock.lock();
			if(0 == buffer.size()) {
				List<T> tasks = taskDao.listTasks(storeSize);
				buffer.addAll(tasks);
			}
		}finally {
			lock.unlock();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#size()
	 */
	@Override
	public int size() {

		return queueSize.get();
	}

	public int getBufferSize() {

		return bufferSize;
	}

	public TaskDao getTaskDao() {

		return taskDao;
	}

	public void setTaskDao(TaskDao taskDao) {

		this.taskDao = taskDao;
	}

	public int getStoreSize() {

		return storeSize;
	}

	private class StoreTask implements Runnable {

		/*
		 * (non-Javadoc)
		 * 
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {

			List<T> tasks = new ArrayList<T>();
			int size = buffer.drainTo(tasks, storeSize);
			if(0 != size) {
				taskDao.batchAddTasks(tasks);
			}
		}

	}
}
